package com.wyx.springbootkafkaconsumerone.service;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import javax.annotation.Resource;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

/**
 * @author wyx
 * @version V 1
 * @Date 2022/11/3
 */
//@Service
public class BsjConsumerService {

  @Resource
  private KafkaTemplate<String, String> kafkaTemplate;


//  @KafkaListener(topics = "smamo-miniapp-pkg-created-dev")
//  public void licecnsePlatListener(ConsumerRecord<String, String> consumerRecord) throws InterruptedException {
//    System.out.println(consumerRecord.toString());
//    System.out.println(consumerRecord);
//    MiniAppPkgCreatedResult obj = JSONObject.parseObject(consumerRecord.value(), MiniAppPkgCreatedResult.class);
//    System.out.println(obj.toString());
//  }

  @KafkaListener(topics = "smamo-miniapp-pkg-created-staging")
  public void licecnsePlatListenerStaging(ConsumerRecord<String, String> consumerRecord) throws InterruptedException {
    System.out.println(consumerRecord.toString());
    System.out.println(consumerRecord);
    MiniAppPkgCreatedResult obj = JSONObject.parseObject(consumerRecord.value(), MiniAppPkgCreatedResult.class);
    System.out.println(obj.toString());
  }

  @KafkaListener(topics = "smamo-miniapp-pkg-created-preprod")
  public void licecnsePlatListenerPP(ConsumerRecord<String, String> consumerRecord) throws InterruptedException {
    System.out.println(consumerRecord.toString());
    System.out.println(consumerRecord);
    MiniAppPkgCreatedResult obj = JSONObject.parseObject(consumerRecord.value(), MiniAppPkgCreatedResult.class);
    System.out.println(obj.toString());
  }


  public void pushMiniAppPkgCreatedResult(String topic, MiniAppPkgCreatedResult param) {
    try {
      Map<String, Object> params = new ConcurrentHashMap<>();
      params.put("uuid", UUID.randomUUID().toString());
      params.put("orderId", param.getOrderId());
      params.put("externalOrderId", param.getExternalOrderId());
      params.put("paymentNo", param.getPaymentNo());
      params.put("packageStatus", param.getPackageStatus());
      params.put("packageId", param.getPackageId());
      params.put("validTo", param.getValidTo());
      params.put("packageActiveTime", param.getPackageActiveTime()); /* 消息发送时间*/
      params.put("msgCreateTime", LocalDateTime.now().toInstant(ZoneOffset.of("Z")).toEpochMilli());
      kafkaTemplate.send(topic, JSONArray.toJSONString(params));

      System.out.println("Kafka message of package send to OMP : " + params);
    } catch (Exception e) {
      System.out.println("pushMiniAppPkgCreatedResult-error" + e);
    }
  }

  public static class MiniAppPkgCreatedResult {
    /**
     * 随机ID， kafka消息投递唯一编号
     */
    private String uuid;
    /**
     * SMAMO 订单号
     */
    private String orderId;
    /**
     * 外部订单号
     */
    private String externalOrderId;
    /**
     * ECF支付单号
     */
    private String paymentNo;
    /**
     * 服务包状态（例：0-未使用）
     */
    private Integer packageStatus;
    /**
     * 服务包ID
     */
    private String packageId;
    /**
     * 服务包到期时间（例：1164691264437）
     */
    private Long validTo;
    /**
     * 服务包生效时间（例：1164691264437）
     */
    private Long packageActiveTime;
    /**
     * 消息发送时间（例：1164691264437）
     */
    private Long msgCreateTime;

    public String getUuid() {
      return uuid;
    }

    public void setUuid(String uuid) {
      this.uuid = uuid;
    }

    public String getOrderId() {
      return orderId;
    }

    public void setOrderId(String orderId) {
      this.orderId = orderId;
    }

    public String getPaymentNo() {
      return paymentNo;
    }

    public void setPaymentNo(String paymentNo) {
      this.paymentNo = paymentNo;
    }

    public Integer getPackageStatus() {
      return packageStatus;
    }

    public void setPackageStatus(Integer packageStatus) {
      this.packageStatus = packageStatus;
    }

    public String getPackageId() {
      return packageId;
    }

    public void setPackageId(String packageId) {
      this.packageId = packageId;
    }

    public Long getValidTo() {
      return validTo;
    }

    public void setValidTo(Long validTo) {
      this.validTo = validTo;
    }

    public Long getPackageActiveTime() {
      return packageActiveTime;
    }

    public void setPackageActiveTime(Long packageActiveTime) {
      this.packageActiveTime = packageActiveTime;
    }

    public Long getMsgCreateTime() {
      return msgCreateTime;
    }

    public void setMsgCreateTime(Long msgCreateTime) {
      this.msgCreateTime = msgCreateTime;
    }

    public String getExternalOrderId() {
      return externalOrderId;
    }

    public void setExternalOrderId(String externalOrderId) {
      this.externalOrderId = externalOrderId;
    }

    @Override
    public String toString() {
      return "MiniAppPkgCreatedResult{" +
          "uuid='" + uuid + '\'' +
          ", orderId='" + orderId + '\'' +
          ", externalOrderId='" + externalOrderId + '\'' +
          ", paymentNo='" + paymentNo + '\'' +
          ", packageStatus=" + packageStatus +
          ", packageId='" + packageId + '\'' +
          ", validTo=" + validTo +
          ", packageActiveTime=" + packageActiveTime +
          ", msgCreateTime=" + msgCreateTime +
          '}';
    }
  }
}
